package org.zjt.flink.stream;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-10-12 下午5:25
 * @see
 */
public class CountWithTimeoutJob {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Random random = new Random();
        // the source data stream
        DataStream<Tuple2<String, String>> stream = env.addSource(new SourceFunction<Tuple2<String, String>>() {
            @Override
            public void run(SourceContext<Tuple2<String, String>> ctx) throws Exception {
                for (int i = 0; i < 100; i++) {
                    ctx.collect(new Tuple2(String.valueOf(random.nextInt(100)),String.valueOf(random.nextInt(100))));
                }
            }

            @Override
            public void cancel() {

            }
        });

// apply the process function onto a keyed stream
        DataStream<Tuple2<String, Long>> result = stream
                .keyBy(0)
                .process(new MyKeyedProcessFunction());


        result.print();

        env.execute();
    }


    static class MyKeyedProcessFunction extends KeyedProcessFunction {

        @Override
        public void processElement(Object value, Context ctx, Collector out) throws Exception {
            System.out.println(value);
            out.collect(value);

            ctx.timerService().registerEventTimeTimer( 60000);
        }
    }


}
